package com.gitee.melin.bee.core.jdbc.dialect;

import com.gitee.melin.bee.core.jdbc.dialect.impl.CassandraJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.ClickHouseJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.DB2JdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.DaMengJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.DorisJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.ElasticSearchDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.Gbase8aJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.GreenplumJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.HashDataJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.HiveJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.IcebergJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.InceptorJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.KafkaDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.KingbaseESJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.MaxComputeJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.MongodbJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.MySqlJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.OceanBaseMySqlJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.OracleJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.PaimonJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.PostgreSqlJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.RedshiftJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.SapHanaJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.SnowflakeJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.SqlServerJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.StarRocksJdbcDialect;
import com.gitee.melin.bee.core.jdbc.dialect.impl.TidbJdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.CompatibleMode;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.google.common.collect.Maps;
import java.sql.Connection;
import java.util.concurrent.ConcurrentMap;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;

public class JdbcDialectHolder {

    private static final Logger LOGGER = LoggerFactory.getLogger(JdbcDialectHolder.class);

    private static final ConcurrentMap<String, Pair<ConnectionInfo, JdbcDialect>> jdbcDialectCache =
            Maps.newConcurrentMap();

    public static JdbcDialect getJdbcDialect(ConnectionInfo connectionInfo) {
        Assert.hasText(connectionInfo.getCode(), "datasource code can not blank");
        Pair<ConnectionInfo, JdbcDialect> pair = jdbcDialectCache.get(connectionInfo.getCode());
        if (pair == null) {
            synchronized (JdbcDialectHolder.class) {
                JdbcDialect jdbcDialect = buildJdbcDialect(connectionInfo);
                jdbcDialectCache.put(connectionInfo.getCode(), Pair.of(connectionInfo, jdbcDialect));
                return jdbcDialect;
            }
        } else {
            if (pair.getLeft().equals(connectionInfo)) {
                return pair.getRight();
            } else {
                synchronized (JdbcDialectHolder.class) {
                    LOGGER.warn("连接信息更新: {}", connectionInfo);
                    jdbcDialectCache.remove(connectionInfo.getCode());
                    try {
                        pair.getRight().close();
                    } catch (Exception e) {
                        LOGGER.error("连接信息更新，close 失败: " + e.getMessage(), e);
                    }
                    return getJdbcDialect(connectionInfo);
                }
            }
        }
    }

    public static JdbcDialect getJdbcDialect() {
        return buildJdbcDialect(null);
    }

    public static JdbcDialect buildJdbcDialect(ConnectionInfo connectionInfo) {
        DataSourceType dataSourceType = connectionInfo.getDataSourceType();
        CompatibleMode compatibleMode = connectionInfo.getCompatibleMode();

        if (DataSourceType.MYSQL == dataSourceType) {
            return new MySqlJdbcDialect(connectionInfo);
        } else if (DataSourceType.DORIS == dataSourceType) {
            return new DorisJdbcDialect(connectionInfo);
        } else if (DataSourceType.STARROCKS == dataSourceType) {
            return new StarRocksJdbcDialect(connectionInfo);
        } else if (DataSourceType.POSTGRESQL == dataSourceType || DataSourceType.GAUSSDWS == dataSourceType) {
            return new PostgreSqlJdbcDialect(connectionInfo);
        } else if (DataSourceType.ORACLE == dataSourceType) {
            return new OracleJdbcDialect(connectionInfo);
        } else if (DataSourceType.SQLSERVER == dataSourceType) {
            return new SqlServerJdbcDialect(connectionInfo);
        } else if (DataSourceType.DB2 == dataSourceType) {
            return new DB2JdbcDialect(connectionInfo);
        } else if (DataSourceType.HANA == dataSourceType) {
            return new SapHanaJdbcDialect(connectionInfo);
        } else if (DataSourceType.HIVE == dataSourceType) {
            return new HiveJdbcDialect(connectionInfo);
        } else if (DataSourceType.GREENPLUM == dataSourceType) {
            return new GreenplumJdbcDialect(connectionInfo);
        } else if (DataSourceType.HASHDATA == dataSourceType) {
            return new HashDataJdbcDialect(connectionInfo);
        } else if (DataSourceType.INCEPTOR == dataSourceType) {
            return new InceptorJdbcDialect(connectionInfo);
        } else if (DataSourceType.DAMENG == dataSourceType) {
            return new DaMengJdbcDialect(connectionInfo);
        } else if (DataSourceType.KINGBASEES == dataSourceType) {
            return new KingbaseESJdbcDialect(connectionInfo);
        } else if (DataSourceType.GBASE8A == dataSourceType) {
            return new Gbase8aJdbcDialect(connectionInfo);
        } else if (DataSourceType.TIDB == dataSourceType) {
            return new TidbJdbcDialect(connectionInfo);
        } else if (DataSourceType.OCEANBASE == dataSourceType) {
            if (compatibleMode == CompatibleMode.MYSQL) {
                return new OceanBaseMySqlJdbcDialect(connectionInfo);
            } else if (compatibleMode == CompatibleMode.ORACLE) {
                return new OceanBaseMySqlJdbcDialect(connectionInfo);
            } else {
                throw new IllegalArgumentException(dataSourceType + " not support compatible mode: " + compatibleMode);
            }
        } else if (DataSourceType.CLICKHOUSE == dataSourceType) {
            return new ClickHouseJdbcDialect(connectionInfo);
        } else if (DataSourceType.CASSANDRA == dataSourceType) {
            return new CassandraJdbcDialect(connectionInfo);
        } else if (DataSourceType.MAXCOMPUTE == dataSourceType) {
            return new MaxComputeJdbcDialect(connectionInfo);
        } else if (DataSourceType.REDSHIFT == dataSourceType) {
            return new RedshiftJdbcDialect(connectionInfo);
        } else if (DataSourceType.SNOWFLAKE == dataSourceType) {
            return new SnowflakeJdbcDialect(connectionInfo);
        } else if (DataSourceType.ELASTICSEARCH == dataSourceType) {
            return new ElasticSearchDialect(connectionInfo);
        } else if (DataSourceType.KAFKA == dataSourceType) {
            return new KafkaDialect(connectionInfo);
        } else if (DataSourceType.MONGODB == dataSourceType) {
            return new MongodbJdbcDialect(connectionInfo);
        } else if (DataSourceType.PAIMON == dataSourceType) {
            return new PaimonJdbcDialect(connectionInfo);
        } else if (DataSourceType.ICEBRG == dataSourceType) {
            return new IcebergJdbcDialect(connectionInfo);
        } else {
            throw new IllegalArgumentException("不支持数据源 " + dataSourceType);
        }
    }

    public static JdbcDialect buildJdbcDialect(DataSourceType dataSourceType, Connection connection) {
        return buildJdbcDialect(dataSourceType, null, connection);
    }

    public static JdbcDialect buildJdbcDialect(
            DataSourceType dataSourceType, CompatibleMode compatibleMode, Connection connection) {
        if (DataSourceType.MYSQL == dataSourceType) {
            return new MySqlJdbcDialect(connection);
        } else if (DataSourceType.DORIS == dataSourceType) {
            return new DorisJdbcDialect(connection);
        } else if (DataSourceType.STARROCKS == dataSourceType) {
            return new StarRocksJdbcDialect(connection);
        } else if (DataSourceType.POSTGRESQL == dataSourceType || DataSourceType.GAUSSDWS == dataSourceType) {
            return new PostgreSqlJdbcDialect(connection);
        } else if (DataSourceType.ORACLE == dataSourceType) {
            return new OracleJdbcDialect(connection);
        } else if (DataSourceType.SQLSERVER == dataSourceType) {
            return new SqlServerJdbcDialect(connection);
        } else if (DataSourceType.DB2 == dataSourceType) {
            return new DB2JdbcDialect(connection);
        } else if (DataSourceType.HANA == dataSourceType) {
            return new SapHanaJdbcDialect(connection);
        } else if (DataSourceType.GREENPLUM == dataSourceType) {
            return new GreenplumJdbcDialect(connection);
        } else if (DataSourceType.HASHDATA == dataSourceType) {
            return new HashDataJdbcDialect(connection);
        } else if (DataSourceType.HIVE == dataSourceType) {
            return new HiveJdbcDialect(connection);
        } else if (DataSourceType.INCEPTOR == dataSourceType) {
            return new InceptorJdbcDialect(connection);
        } else if (DataSourceType.DAMENG == dataSourceType) {
            return new DaMengJdbcDialect(connection);
        } else if (DataSourceType.KINGBASEES == dataSourceType) {
            return new KingbaseESJdbcDialect(connection);
        } else if (DataSourceType.GBASE8A == dataSourceType) {
            return new Gbase8aJdbcDialect(connection);
        } else if (DataSourceType.TIDB == dataSourceType) {
            return new TidbJdbcDialect(connection);
        } else if (DataSourceType.OCEANBASE == dataSourceType) {
            if (compatibleMode == CompatibleMode.MYSQL) {
                return new OceanBaseMySqlJdbcDialect(connection);
            } else if (compatibleMode == CompatibleMode.ORACLE) {
                return new OceanBaseMySqlJdbcDialect(connection);
            } else {
                throw new IllegalArgumentException(dataSourceType + " not support compatible mode: " + compatibleMode);
            }
        } else if (DataSourceType.CLICKHOUSE == dataSourceType) {
            return new ClickHouseJdbcDialect(connection);
        } else if (DataSourceType.CASSANDRA == dataSourceType) {
            return new CassandraJdbcDialect(connection);
        } else if (DataSourceType.REDSHIFT == dataSourceType) {
            return new RedshiftJdbcDialect(connection);
        } else if (DataSourceType.SNOWFLAKE == dataSourceType) {
            return new SnowflakeJdbcDialect(connection);
        } else if (DataSourceType.PAIMON == dataSourceType) {
            return new PaimonJdbcDialect(connection);
        } else if (DataSourceType.ICEBRG == dataSourceType) {
            return new IcebergJdbcDialect(connection);
        } else {
            throw new IllegalArgumentException("不支持数据源 " + dataSourceType);
        }
    }
}
